package com.bw.gmall.app.ods;

import com.google.gson.Gson;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ThreadLocalRandom;

public class KafkaDataGenerator {
    // Kafka 配置参数
    private static final String KAFKA_BOOTSTRAP_SERVERS = "hadoop102:9092";
    private static final String KAFKA_TOPIC = "user_biao_topic";
    private static final int RECORD_COUNT = 100; // 生成的数据条数

    // 时间范围集合
    private static final List<String> TIME_RANGES = Arrays.asList(
        "0-3个月", "3-6个月", "6-12个月", "12-24个月","24-36个月","孕中期（13-28周）","孕晚期（29-40周）",
        "孕早期（0-12周）", "3-6岁", "6-12岁"
    );

    // 行为类型集合
    private static final List<String> BEHAVIOR_TYPES = Arrays.asList(
        "购买", "搜索", "收藏", "浏览"
    );

    // 数据模板
    private static final List<Map<String, String>> BASE_TEMPLATES = Arrays.asList(
        new HashMap<String, String>() {{
            put("product_category", "童装/婴儿装/亲子装 > 女童连衣裙");
            put("product_title", "女童连衣裙夏季公主裙{time_range}宝宝裙子");
            put("product_keyword", "女童，{time_range}，连衣裙");
        }},
        new HashMap<String, String>() {{
            put("product_category", "玩具/童车/益智/积木/模型 > 电动遥控玩具");
            put("product_title", "勇敢电动遥控车{time_range}儿童玩具赛车");
            put("product_keyword", "男童，{time_range}，遥控车");
        }},
        new HashMap<String, String>() {{
            put("product_category", "童装/婴儿装/亲子装 > 连身衣/爬服/哈衣");
            put("product_title", "婴儿连身衣春秋款{time_range}男女通用");
            put("product_keyword", "婴儿，{time_range}，通用");
        }},
        new HashMap<String, String>() {{
            put("product_category", "玩具/童车/益智/积木/模型 > STEAM玩具");
            put("product_title", "儿童STEAM积木套装{time_range}益智玩具");
            put("product_keyword", "STEAM,{time_range}，积木");
        }},
        new HashMap<String, String>() {{
            put("product_category", "童装/婴儿装/亲子装 > 男童T恤");
            put("product_title", "男童短袖T恤夏季{time_range}纯棉上衣");
            put("product_keyword", "男童，{time_range}，T恤");
        }}
    );

    public static void main(String[] args) {
        // 配置 Kafka 生产者
        Properties props = new Properties();
        props.put("bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        Gson gson = new Gson();

        try {
            for (int i = 0; i < RECORD_COUNT; i++) {
                // 生成数据
                Map<String, String> data = generateData();
                
                // 转换为 JSON 并发送到 Kafka
                String jsonData = gson.toJson(data);
                ProducerRecord<String, String> record = new ProducerRecord<>(
                    KAFKA_TOPIC, 
                    data.get("user_id"),  // 使用 user_id 作为 key
                    jsonData
                );
                
                producer.send(record);
                System.out.printf("发送数据: %s%n", jsonData);
            }
            
            // 确保所有消息都被发送
            producer.flush();
            System.out.println("所有数据已成功发送到 Kafka");
        } finally {
            producer.close();
        }
    }

    // 生成单条数据
    private static Map<String, String> generateData() {
        // 随机选择模板和时间范围
        Map<String, String> template = BASE_TEMPLATES.get(
            ThreadLocalRandom.current().nextInt(BASE_TEMPLATES.size())
        );
        String timeRange = TIME_RANGES.get(
            ThreadLocalRandom.current().nextInt(TIME_RANGES.size())
        );
        String behaviorType = BEHAVIOR_TYPES.get(
            ThreadLocalRandom.current().nextInt(BEHAVIOR_TYPES.size())
        );

        // 构建数据对象
        Map<String, String> data = new HashMap<>();
        data.put("user_id", generateUserId());
        data.put("behavior_type", behaviorType);
        data.put("behavior_time", generateBehaviorTime());
        data.put("product_category", template.get("product_category"));
        data.put("product_title", template.get("product_title").replace("{time_range}", timeRange));
        data.put("product_keyword", template.get("product_keyword").replace("{time_range}", timeRange));

        return data;
    }

    // 生成随机用户ID
    private static String generateUserId() {
        return "U" + (000 + ThreadLocalRandom.current().nextInt(1000));
    }

    // 生成随机行为时间（近60天内）
    private static String generateBehaviorTime() {
        LocalDateTime now = LocalDateTime.now();
        int daysAgo = ThreadLocalRandom.current().nextInt(61); // 0-60天前
        LocalDateTime behaviorTime = now.minusDays(daysAgo)
            .withHour(ThreadLocalRandom.current().nextInt(24))
            .withMinute(ThreadLocalRandom.current().nextInt(60))
            .withSecond(ThreadLocalRandom.current().nextInt(60));
        
        return behaviorTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
    }
}